package com.gmall.app.function;

import com.alibaba.fastjson.JSONObject;
import com.gmall.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws SQLException, ClassNotFoundException {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        connection.setAutoCommit(true);
    }

    @Override
    public void invoke(JSONObject value,Context ctx) throws SQLException {
        PreparedStatement preparedStatement = null;
        try{
            //获取SQL语句
            String upsertSql = genUpsertSql(value.getString("sinkTable"),
                    value.getJSONObject("after"));
            System.out.println(upsertSql);

            //预编译sql
            PreparedStatement prepareStatement = connection.prepareStatement(upsertSql);
            prepareStatement.executeUpdate();

        }catch (SQLException e){
            e.printStackTrace();
        }finally {
            if (preparedStatement != null){
                preparedStatement.close();
            }
        }
    }

    private String genUpsertSql(String sinkTable, JSONObject data) {
        Set<String> keySet = data.keySet();
        Collection<Object> values = data.values();

        return "upsert into "+GmallConfig.HBASE_SCHEMA + "," + sinkTable + "(" +
                StringUtils.join(keySet,",")+") values('" +
                StringUtils.join(values,"','")+"')";

    }
}
